package cn.doitedu.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MD5Hash;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;

public class GetDemo {

    public static void main(String[] args) throws IOException {

        Configuration conf = HBaseConfiguration.create();
        Connection conn = ConnectionFactory.createConnection(conf);
        Table table = conn.getTable(TableName.valueOf("doit:students"));

        // getKeyValue(table);
        // getFamily(table);
        // scanRows(table);
        scanRowsWithFilter(table);

    }

    /**
     * 取一个指定的KeyValue
     * @param table
     */
    public static void getKeyValue(Table table) throws IOException {


        String rowKey = MD5Hash.getMD5AsHex(Bytes.toBytes("000010"));

        Get getParam = new Get(Bytes.toBytes(rowKey));
        getParam.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("age"));


        Result result = table.get(getParam);

        byte[] value = result.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("age"));

        int age = Bytes.toInt(value);

        System.out.println(age);

        table.close();

    }


    /**
     * get 单行中的指定列族的所有KeyValue数据
     * @param table
     * @throws IOException
     */
    public static void getFamily(Table table) throws IOException {

        String rowKey = MD5Hash.getMD5AsHex(Bytes.toBytes("000010"));

        Get getParam = new Get(Bytes.toBytes(rowKey));
        getParam.addFamily(Bytes.toBytes("base_info"));

        Result result = table.get(getParam);
        CellScanner cellScanner = result.cellScanner();
        while(cellScanner.advance()){
            // 一个cell就代表一个KeyValue
            Cell cell = cellScanner.current();

            byte[] rowKeyBytes = CellUtil.cloneRow(cell);
            byte[] familyBytes = CellUtil.cloneFamily(cell);
            byte[] qualifierBytes = CellUtil.cloneQualifier(cell);
            byte[] valueBytes = CellUtil.cloneValue(cell);

            String qualifier = Bytes.toString(qualifierBytes);
            String value = "";
            if("age".equals(qualifier)){
                value = Bytes.toInt(valueBytes)+"";
            }else{
                value =  Bytes.toString(valueBytes);
            }

            System.out.println(Bytes.toString(rowKeyBytes) + ","
              + Bytes.toString(familyBytes) + ","
              + qualifier + ","
              + value
            );

        }

        table.close();

    }


    /**
     * 扫描指定行范围的数据
     * @param table
     * @throws IOException
     */
    public static void scanRows(Table table) throws IOException {

        Scan scan = new Scan();
        String startMd5 = MD5Hash.getMD5AsHex(Bytes.toBytes("001000"));
        String stopMd5 = MD5Hash.getMD5AsHex(Bytes.toBytes("002000"));
        // 设置了扫描的起始行和结束行（默认含首不含尾）
        scan.withStartRow(Bytes.toBytes(startMd5));
        //scan.withStopRow(Bytes.toBytes(stopMd5),false);

        // 设置总共要扫描的行数
        scan.setLimit(10);

        ResultScanner scanner = table.getScanner(scan);
        // 扫描时的  行 迭代器
        Iterator<Result> iterator = scanner.iterator();
        printData(iterator);

        table.close();

    }


    /**
     * 扫描多行数据，并带过滤条件
     * @param table
     */
    public static void scanRowsWithFilter(Table table) throws IOException {
        Scan scan = new Scan();

        scan.withStartRow(Bytes.toBytes("eccb22fe1acf4027649239c7d6383d78"),true);
        scan.setLimit(10);

        // 行键前缀过滤器
        Filter filter1 = new PrefixFilter(Bytes.toBytes("ecc"));  // 行键前缀过滤器

        // 列值过滤器（匹配到条件的行，整行数据都将返回）
        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("extra_info"), Bytes.toBytes("city"), CompareOperator.EQUAL, Bytes.toBytes("南京"));

        ArrayList<Filter> filterList = new ArrayList<>();
        filterList.add(filter1);
        filterList.add(filter2);

        // 将上面的2个过滤器，组成一个 MUST_PASS_ALL 关系的过滤器组
        FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL,filterList);
        //FilterListWithAND filters = new FilterListWithAND(filterList);


        // 给scan参数设置过滤器条件
        scan.setFilter(filters);


        ResultScanner scanner = table.getScanner(scan);
        Iterator<Result> iterator = scanner.iterator();  // 拿到行迭代器

        printData(iterator);  // 迭代，并打印数据

        table.close();
    }


    /**
     * 用于打印 scan结果的工具方法
     * @param iterator
     * @throws IOException
     */
    private static void printData(Iterator<Result> iterator) throws IOException {
        while(iterator.hasNext()){  // 行迭代
            // 迭代到一行
            Result result = iterator.next();
            // 拿到行中的cell单元格的迭代器
            CellScanner cellScanner = result.cellScanner();
            // 用单元格迭代器，迭代行中的每一个cell（单元格，KeyValue）
            while(cellScanner.advance()){   //  行内的keyValue迭代
                Cell cell = cellScanner.current();

                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = "";
                if(qualifier.equals("age") || qualifier.equals("salary")){
                    value = Bytes.toInt(CellUtil.cloneValue(cell))+"";
                }else{
                    value = Bytes.toString(CellUtil.cloneValue(cell));
                }

                System.out.println(rowKey+","+family+","+qualifier + ","+value);

            }
        }
    }


}
